package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo02SourceAPI {
  def main(args: Array[String]): Unit = {
    /**
     * Spark中常见的Source：
     * csv、json、parquet、orc、jdbc等
     */

    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo02SourceAPI")
      .master("local")
      .getOrCreate()

    import spark.implicits._


    // 1、csv
    val stuDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id String,name String,age Int,gender String,clazz String")
      .load("spark/data/students.txt")

    //    stuDF.show()

    stuDF
      //      .where("age>23") // 字符串表达式
      .where($"age" > 23) // 列表达式
      .select($"id", $"name", $"age", $"clazz")
    //      .show()

    // 指定格式保存数据
    stuDF
      .write
      .format("json")

      /**
       * 写入时可以指定保存模式SaveMode
       * 总共有四种模式：
       * Append追加
       * Ignore存在即跳过
       * Overwrite覆盖写
       * ErrorIfExists存在即报错
       */
      .mode(SaveMode.Overwrite)
    //      .save("spark/data/stu/json")

    stuDF.write.format("parquet").mode(SaveMode.Overwrite)
    //      .save("spark/data/stu/parquet")
    stuDF.write.format("orc").mode(SaveMode.Overwrite)
    //      .save("spark/data/stu/orc")

    // 2、json格式
    val jsonStuDF: DataFrame = spark
      .read
      .format("json")
      //      .schema() // json格式自带结构，不需要指定schema
      .load("spark/data/stu/json")

    //    jsonStuDF.show()

    // 3、parquet
    // parquet格式自带结构，不需要指定schema，
    // 无法直接打开查看数据，它是一种二进制的格式
    // 默认会使用snappy的方式对数据进行压缩，适合存嵌套或者结构化的数据
    val parquetStuDF: DataFrame = spark
      .read
      .format("parquet")
      //      .schema()
      .load("spark/data/stu/parquet")

    //    parquetStuDF.show()

    // 4、orc
    /**
     * orc格式自带结构，不需要指定schema，
     * 无法直接打开查看数据，它是一种二进制的格式
     * 默认会使用snappy的方式对数据进行压缩，文件空间占用率最小
     */
    val orcStuDF: DataFrame = spark
      .read
      .format("orc")
      //      .schema()
      .load("spark/data/stu/orc")

    //    orcStuDF.show()

    // 5、JDBC
    val jdbcDF: DataFrame = spark
      .read
      .format("jdbc")
      .option("url", "jdbc:mysql://rm-bp1h7v927zia3t8iwho.mysql.rds.aliyuncs.com:3307/city_crm_21")
      .option("dbtable", "students")
      .option("user", "shujia21")
      .option("password", "123456")
      .load()

    jdbcDF.show()

    // 将DF写入MySQL
    jdbcDF.write
      .format("jdbc")
      .option("url", "jdbc:mysql://rm-bp1h7v927zia3t8iwho.mysql.rds.aliyuncs.com:3307/city_crm_21")
      .option("dbtable", "students_spark_sql")
      .option("user", "shujia21")
      .option("password", "123456")
      .option("truncate", "true") // 使用truncate的方式清理数据 再实现overwrite 防止表结构发生变更
      .mode(SaveMode.Overwrite) // 默认会先将表drop掉再重新create，会导致表结构发生变化，String类型的值会用text类型接受
      .save()


  }

}
